package com.example.mq.mqserver.datacenter;

import com.example.mq.common.BinaryTool;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.MSGQueue;
import com.example.mq.mqserver.core.Message;

import java.io.*;
import java.util.LinkedList;
import java.util.Scanner;

/**
 * 通过这个类,来针对硬盘上的消息进行管理
 */
public class MessageFileManager {
    public void init() {
        // TODO 啥也不做, 以备后用
    }

    // 定义一个内部类, 来表示队列的统计信息
    // 优先考虑使用 static , 静态内部类
    static public class Stat {
        // 此处直接定义成 public, 就不必写 get, set 方法了
        // 对于这样简单的类, 就直接操作成员, 类似于 C 中结构体
        public int totalCount; // 总消息数量
        public int validCount; //有效消息数量
    }

    // 预定消息文件所在的目录和文件名
    // 获取到执行队列对应的消息文件所在的路径
    private String getQueueDir(String queueName) {
        return "./data/" + queueName;
    }

    // 获取队列的消息数据文件路径
    // 二进制文件一般使用 .bin / .dat 作为后缀, 而不是使用 .txt, 此处不做太多考虑
    private String getQueueDataPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data.txt";
    }

    // 获取队列的消息统计文件路径
    private String getQueueStatPath(String queueName) {
        return getQueueDir(queueName) + "/queue_stat.txt";
    }

    /**
     * 获取消息统计文件中内容
     * @param queueName
     * @return
     */
    private Stat readStat(String queueName) {
        // 由于当前的消息统计文件是文本文件, 可以直接用 Scanner 来读取数据 "2000\t1500"
        Stat stat = new Stat();
        try(InputStream inputStream = new FileInputStream(getQueueStatPath(queueName))) {
            Scanner scanner = new Scanner(inputStream);
            stat.totalCount = scanner.nextInt();
            stat.validCount = scanner.nextInt();
            return stat;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 往消息统计文件中写数据
     * @param queueName
     * @param stat
     */
    private void writeStat(String queueName, Stat stat) {
        // 使用 PrintWriter 来写文件
        // append 属性默认为 false-不追加, 即覆盖原数据
        try(OutputStream outputStream = new FileOutputStream(getQueueStatPath(queueName), false)) {
            PrintWriter printWriter = new PrintWriter(outputStream);
            printWriter.write(stat.totalCount + "\t" + stat.validCount);
            printWriter.flush();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 创建队列对应的文件和目录
     * @param queueName
     * @throws IOException
     */
    public void createQueueFiles(String queueName) throws IOException {
        // 1. 先创建队列对应的消息目录
        File baseDir = new File(getQueueDir(queueName));
        if(!baseDir.exists()) {
            // 不存在则创建目录
            boolean ok = baseDir.mkdirs();
            if(!ok) {
                throw new IOException("创建目录失败! baseDir=" + baseDir.getAbsolutePath());
            }
        }

        // 2.创建队列数据文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        if(!queueDataFile.exists()) {
            boolean ok = queueDataFile.createNewFile();
            if(!ok) {
                throw new IOException("创建文件失败! queueDataFile=" + queueDataFile.getAbsolutePath());
            }
        }

        // 3.创建消息统计文件
        File queueStatFile = new File(getQueueStatPath(queueName));
        if(!queueStatFile.exists()) {
            boolean ok = queueStatFile.createNewFile();
            if(!ok) {
                throw new IOException("创建文件失败! queueStatFile=" + queueStatFile.getAbsolutePath());
            }
        }

        // 4. 给消息统计文件设定初始值 -- "0\t0"
        Stat stat = new Stat();
        stat.totalCount = 0;
        stat.validCount = 0;
        writeStat(queueName, stat);
    }

    /**
     * 删除队列的目录和文件
     * 当队列删除的时候, 该队列对应的目录和文件也要删除
     * @param queueName
     * @throws IOException
     */
    public void destroyQueueFiles(String queueName) throws IOException {
        // 先删除文件, 再删除目录
        File queueStatFile = new File(getQueueStatPath(queueName));
        boolean ok1 = queueStatFile.delete();
        File queueDataFile = new File(getQueueDataPath(queueName));
        boolean ok2 = queueDataFile.delete();
        File baseDir = new File(getQueueDir(queueName));
        boolean ok3 = baseDir.delete();
        if(!(ok1 && ok2 && ok3)) {
            // 文件和目录有一个删除失败, 都算整体删除失败
            throw new IOException("删除队列和文件失败 baseDir=" + baseDir.getAbsolutePath());
        }
    }

    /**
     * 检查队列的目录和文件是否存在
     * @param queueName
     * @return
     */
    public boolean checkFilesExits(String queueName) {
        // 文件都存在, ,则目录一定存在
        File queueDataFile = new File(getQueueDataPath(queueName));
        if(!queueDataFile.exists()) {
            return false;
        }
        File queueStatFile = new File(getQueueStatPath(queueName));
        if(!queueStatFile.exists()) {
            return false;
        }
        return true;
    }

    /**
     * 添加消息
     * 该方法用于把一个消息, 放到对应的队列中
     * @param queue
     * @param message
     * @throws MqException
     * @throws IOException
     */
    public void sendMessage(MSGQueue queue, Message message) throws MqException, IOException {
        // 1.检查当前要写入的队列对应的文件是否存在
        if(!checkFilesExits(queue.getName())) {
            throw new MqException("[MessageFileManager] 队列对应的文件不存在! queueName=" + queue.getName());
        }
        // 2.把 Message 对象, 进行序列化, 转成二进制的字节数组
        byte[] messageBinary = BinaryTool.toByte(message);

        // 对 消息队列 加锁, 即对每一个 队列对应的资源和操作加锁
        synchronized (queue) {
            // 3.获取当前队列 数据文件 的长度, 以此计算出 Message 对象的 offsetBeg 和 offsetEnd
            // 把新的 Message 数据, 写入到队列的末尾
            // offsetBeg: 当前文件长度 + 4 (四字节放 Message 长度)
            // offsetEnd: 当前文件长度 + 4 + Message自身长度
            File queueDataFile = new File(getQueueDataPath(queue.getName()));
            // queueDataFile.length() 可以获取到文件的长度, 单位字节.
            message.setOffsetBeg(queueDataFile.length() + 4);
            message.setOffsetEnd(queueDataFile.length() + 4 + messageBinary.length);

            // 4.写入消息到数据文件 (此时是追加写)
            try (OutputStream outputStream = new FileOutputStream(queueDataFile, true)) {
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    // outputStream.write() 是一次写一个字节, 可以分四次写四个字节
                    // 也可以用 DataOutputStream.writeInt() 直接写四个字节的内容
                    dataOutputStream.writeInt(messageBinary.length);
                    dataOutputStream.write(messageBinary);
                }
            }

            // 5.更新消息统计文件内容
            Stat stat = readStat(queue.getName());
            stat.totalCount++;
            stat.validCount++;
            writeStat(queue.getName(), stat);
        }
    }

    /**
     * 删除消息 (删除某个队列的某个消息)
     * 逻辑删除, 把该消息对应的那个 isValid 属性, 设置成 0
     * 1.先把消息读出来,反序列化成 Message 对象
     * 2.将 isValid 属性设置为 0x0
     * 3.把消息序列化后写回文件
     * @param queue 需要通过 queue, 来获取该队列对应的文件: queue_data.txt
     * @param message 主要是使用 message 的一些属性: offsetBeg, offsetEnd
     */
    public void deleteMessage(MSGQueue queue, Message message) throws IOException, ClassNotFoundException {
        // 对 queue 对象进行加锁, 多个 队列 之间操作互不影响, 对于同一个队列删除的时候, 不能进行其他操作 (脏数据)
        synchronized (queue) {
            // 内存中 data文件逻辑删除
            try(RandomAccessFile randomAccessFile = new RandomAccessFile(getQueueDataPath(queue.getName()), "rw")) {
                // 1.先从文件中读出对应的 Message 数据
                byte[] bufferSrc = new byte[(int) (message.getOffsetEnd()-message.getOffsetBeg())];
                // 先定位光标, 在进行读操作
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.read(bufferSrc); //输出型参数
                // 2.把当前读出的二进制数据, 转换成 Message 数据 (反序列化)
                Message diskMessage = (Message) BinaryTool.fromByte(bufferSrc);
                // 3.把 isValid 设置为无效
                diskMessage.setIsValid((byte) 0x0);
                // 4.重新写入文件
                byte[] bufferDest = BinaryTool.toByte(diskMessage);
                // 随着读操作, 光标会移动, 此时需要重新定位光标
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.write(bufferDest);
            }
            // 更新统计文件
            Stat stat = readStat(queue.getName());
            if(stat.validCount > 0) {
                stat.validCount--;
            }
            writeStat(queue.getName(), stat);
        }
    }


    /**
     * 使用该方法从文件中读取出所有的消息内容,加载到内存中 (放到一个链表这样的结构里)
     * 该方法会在程序启动的时候调用, 此时服务器还不能处理请求, 因此不涉及多线程操作, 也就不需要加锁 (使用 String, 而不是使用 对象)
     * @param queueName
     * @return
     */
    public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws IOException, MqException, ClassNotFoundException {
        LinkedList<Message> messages = new LinkedList<>();
        try(InputStream inputStream = new FileInputStream(getQueueDataPath(queueName))) {
            try(DataInputStream dataInputStream = new DataInputStream(inputStream)) {
                // 使用该变量记录文件光标
                long currentOffset = 0;
                // 一个队列中有多条消息, 循环读取所有的信息
                while(true) {
                    // 1.读取当前消息长度
                    int messageSize = dataInputStream.readInt();
                    // 2.读取当前消息内容
                    byte[] buffer = new byte[messageSize];
                    int actualSize = dataInputStream.read(buffer);
                    if(messageSize != actualSize) {
                        // 如果预期读取长度和实际读取长度不一致, 则说明文件有问题
                        throw new MqException("[MessageFileManager] 文件格式错误! queueName=" + queueName);
                    }
                    // 3.将读到的二进制数据反序列化成 Message 对象
                    Message message = (Message) BinaryTool.fromByte(buffer);
                    // 4.判定该消息对象是否是无效数据
                    if(message.getIsValid() != 0x1) {
                        // 无效数据, 跳过即可
                        // 但是文件光标还需要进行更新
                        currentOffset += (4 + messageSize);
                        continue;
                    }
                    // 5.有效数据, 需要在将 message 添加到列表中之前, 填写 offsetBeg 和 offsetEnd
                    message.setOffsetBeg(currentOffset + 4);
                    message.setOffsetEnd(currentOffset + 4 + messageSize);
                    currentOffset += (4 + messageSize);

                    messages.add(message);
                }
            } catch (EOFException e) {
                // DataInputStream.readInt() 读取到文件末尾, 就会触发 EOFException 异常
                // 所以这里的异常属于正常业务逻辑
                System.out.println("[MessageFileManager] 恢复 Message 数据完成!");
            }
        }
        return messages;
    }

    /**
     * 判断当前队列是否需要进行 GC
     * @param queueName
     * @return
     */
    public boolean checkGC(String queueName) {
        // 从统计文件中获取 totalCount 和 validCount
        Stat stat = readStat(queueName);
        if((stat.totalCount > 2000) && (stat.totalCount > stat.validCount * 2)) {
            return true;
        }
        return false;
    }

    private String getQueueDataNewPath(String queueName) {
        // ./data/testQueue/queue_data_new.txt
        return getQueueDir(queueName) + "/queue_data_new.txt";
    }

    /**
     * 该方法用来执行 消息数据文件 的 垃圾回收操作.
     * 使用复制算法完成
     * 1.创建一个新文件--queue_data_new.txt
     * 2.将有效数据读取到新文件中
     * 3.删除旧数据, 将新文件的名字改成 queue_data.txt
     * 4.更新统计信息
     * @param queue
     */
    public void gc(MSGQueue queue) throws MqException, IOException, ClassNotFoundException {
        // 进行 gc 的时候, 其他增删查改操作不准执行
        synchronized (queue) {
            // 由于 gc 操作比较耗时, 可打印一下执行 gc 的时长
            long gcBeg = System.currentTimeMillis();

            // 1.创建一个新文件--queue_data_new.txt
            File queueDataNewFile = new File(getQueueDataNewPath(queue.getName()));
            if(queueDataNewFile.exists()) {
                // 正常情况下, 该文件不应该存在 (存在则说明之前的 gc 操作完成了一半就异常退出了?)
                throw new MqException("[MessageFileManager] gc 时发现该队列的 queue_data_new.txt 文件已存在! queueName=" + queue.getName());
            }
            boolean ok = queueDataNewFile.createNewFile();
            if(!ok) {
                throw new MqException("[MessageFileManager] gc 时创建 queue_data_new.txt 文件失败! queueDataNewFile=" + queueDataNewFile.getAbsolutePath());
            }

            // 2.从旧的文件中, 将有效数据读取到新文件中
            LinkedList<Message> messages = loadAllMessageFromQueue(queue.getName());

            // 3.把有效消息, 写入到新的文件中
            try(OutputStream outputStream = new FileOutputStream(queueDataNewFile)) {
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)){
                    for (Message message : messages) {
                        byte[] buffer = BinaryTool.toByte(message);
                        // 写消息长度
                        dataOutputStream.writeInt(buffer.length);
                        // 写消息内容
                        dataOutputStream.write(buffer);
                    }
                }
            }

            // 4.删除旧数据, 将新文件的名字改成 queue_data.txt
            File queueDataOldFile = new File(getQueueDataPath(queue.getName()));
            ok = queueDataOldFile.delete();
            if(!ok) {
                throw new MqException("[MessageFileManager] gc, 删除旧的数据文件失败! queueDataOldFile=" + queueDataOldFile.getAbsolutePath());
            }
            // 重命名: queue_data_new.txt => queue_data.txt
            ok = queueDataNewFile.renameTo(queueDataOldFile);
            if(!ok) {
                throw new MqException("[MessageFileManager] 文件重命名失败! queueDataNewFile=" + queueDataNewFile.getAbsolutePath()
                + ", queueDataOldFile=" + queueDataOldFile.getAbsolutePath());
            }

            // 5.更新统计信息
            Stat stat = readStat(queue.getName());
            stat.totalCount = messages.size();
            stat.validCount = messages.size();
            writeStat(queue.getName(), stat);



            long gcEnd = System.currentTimeMillis();
            System.out.println("[MessageFileManager] gc 执行完毕! queueName=" + queue.getName() + ", time="
                    + (gcEnd - gcBeg) + "ms");
        }
    }
}
